package fun.tan90.easy.binlog.worker.handler.listener;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import fun.tan90.easy.binlog.base.threadpool.EasyBinlogThreadPool;
import fun.tan90.easy.binlog.worker.handler.processor.BinlogEventProcessor;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/9/1 12:53
 */
@Slf4j
public class BinlogLifecycleListener implements BinaryLogClient.LifecycleListener {

    private final static Map<String, ScheduledFuture<?>> SCHEDULED_FUTURE_MAP = new ConcurrentHashMap<>();

    private final String masterCode;

    private final BinlogEventProcessor binlogEventProcessor;

    public BinlogLifecycleListener(String masterCode, BinlogEventProcessor binlogEventProcessor) {
        this.masterCode = masterCode;
        this.binlogEventProcessor = binlogEventProcessor;
    }

    @Override
    public void onConnect(BinaryLogClient binaryLogClient) {
        ScheduledFuture<?> scheduledFuture = EasyBinlogThreadPool.EB_SCHEDULED_EXECUTOR.scheduleAtFixedRate(() -> {
            String binlogFilename = binaryLogClient.getBinlogFilename();
            long binlogPosition = binaryLogClient.getBinlogPosition();
            log.info("binlogFilename {}, binlogPosition {}", binlogFilename, binlogPosition);
            try {
                binlogEventProcessor.rotateBinlogFile(masterCode, binlogFilename, binlogPosition);
            } catch (Exception e) {
                log.error(e.getMessage());
            }
        }, 1, 5, TimeUnit.SECONDS);
        SCHEDULED_FUTURE_MAP.put(masterCode, scheduledFuture);
    }

    @Override
    public void onCommunicationFailure(BinaryLogClient binaryLogClient, Exception e) {
        log.error("onCommunicationFailure {}, error {}", binaryLogClient.getServerId(), e.getMessage());
    }

    @Override
    public void onEventDeserializationFailure(BinaryLogClient binaryLogClient, Exception e) {
        log.error("onEventDeserializationFailure {}, error {}", binaryLogClient.getServerId(), e.getMessage());
    }

    @Override
    public void onDisconnect(BinaryLogClient binaryLogClient) {
        ScheduledFuture<?> scheduledFuture = SCHEDULED_FUTURE_MAP.get(masterCode);
        if (!Objects.isNull(scheduledFuture) && !scheduledFuture.isCancelled()) {
            scheduledFuture.cancel(true);
        }
    }

}
